package streaming.api.transform;

import streaming.api.beans.SensorReading;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 滚动聚合
 * KeyBy
 * sum()
 * min()
 * max()
 * minBy()
 * maxBy()
 */
public class TransformTest2 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStream<String> inputStream = env.readTextFile("D:\\IdeaProjects\\springboot-flink-1\\flinkTutorial\\src\\main\\resources\\sensor.txt");

        DataStream<SensorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0],new Long(fields[1]), new Double(fields[2]));
        });

        //分组
        //KeyedStream<SensorReading, Tuple> keyedStream = dataStream.keyBy("id");
        //KeyedStream<SensorReading, String> keyedStream = dataStream.keyBy(data -> data.getId());
        KeyedStream<SensorReading, String> keyedStream = dataStream.keyBy(SensorReading::getId);
        //max，取当前最大的温度值
        DataStream<SensorReading> result1 = keyedStream.max("temperature");
        //maxBy，取当前最大的温度值
        DataStream<SensorReading> result2 = keyedStream.maxBy("temperature");
        //min，取当前最小的时间戳
        DataStream<SensorReading> result3 = keyedStream.min("timestamp");
        //sum，统计所有时间戳
        DataStream<SensorReading> result4 = keyedStream.sum("timestamp");
        //result1.print("max:");
        //result2.print("maxBy:");
        //result3.print("min:");
        result4.print("sum:");
        env.execute();
    }
}
